package com.spring.zabelection.websocket;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.spring.zabelection.service.LeaderElectionService;
import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.*;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.client.WebSocketConnectionManager;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.net.URI;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class WebSocketClientService {


    @Value("${websocket.peers}")
    private String peerUrlsConfig;

    private List<String> peerUrls;
    private final Map<String, WebSocketSession> activeSessions = new ConcurrentHashMap<>();
    private final Map<String, Integer> retryCounts = new ConcurrentHashMap<>();
    private static final int MAX_RETRY = 5;

    private final ObjectMapper objectMapper = new ObjectMapper();

    @Autowired
    private LeaderElectionService leaderElectionService;

    @PostConstruct
    public void initialize() {
        // 解析 peerUrlsConfig（因为 @Value 不能直接注入 List）
        peerUrls = List.of(peerUrlsConfig.split(","))
                .stream()
                .map(String::trim)
                .collect(Collectors.toList());
        connectToPeers();
    }

    // 定期检查连接状态，如果未连接则重试
    @Scheduled(fixedDelay = 5000)
    public void retryConnections() {
        for (String url : peerUrls) {
            if (!activeSessions.containsKey(url)) {
                System.out.println("🔄 Retrying connection to " + url);
                connectToPeer(url);
            }
        }
    }

    private void connectToPeers() {
        for (String url : peerUrls) {
            connectToPeer(url);
        }
    }

    private void connectToPeer(String url) {
        WebSocketClient client = new StandardWebSocketClient();

        try {
            WebSocketConnectionManager connectionManager = new WebSocketConnectionManager(client, new TextWebSocketHandler() {
                @Override
                public void afterConnectionEstablished(WebSocketSession session) {
                    activeSessions.put(url, session);
                    retryCounts.put(url, 0);
                    System.out.println("✅ Successfully connected to " + url);


                    try {
                        Thread.sleep(1000 * 10);
                        System.out.println("10秒后开始投票查看效果~");
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    // 连接成功后，立即发送当前节点的投票信息
                    sendVote(session);
                }

                @Override
                protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
                    System.out.println("📩 客户端收到信息 from " + url + ": " + message.getPayload());
                    var vote = objectMapper.readValue(message.getPayload(), WebSocketServer.Vote.class);
                    leaderElectionService.receiveVote(vote.getServerId(), vote.getZxid());
                }

                @Override
                public void handleTransportError(WebSocketSession session, Throwable exception) {
                    System.err.println("❌ Connection error with " + url + ": " + exception.getMessage());
                    closeSession(url);
                }
            }, URI.create(url));

            connectionManager.start();
        } catch (Exception e) {
            int retry = retryCounts.getOrDefault(url, 0);
            if (retry < MAX_RETRY) {
                retryCounts.put(url, retry + 1);
                int waitTime = (int) Math.pow(2, retry); // 指数退避
                System.out.println("⏳ Connection failed, retrying in " + waitTime + "s: " + url);
                try {
                    TimeUnit.SECONDS.sleep(waitTime);
                    connectToPeer(url); // 递归重试
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            } else {
                System.err.println("🚫 Max retries reached for " + url + ", giving up.");
            }
        }
    }

    private void sendVote(WebSocketSession session) {
        try {
            String message = objectMapper.writeValueAsString(Map.of(
                    "serverId", leaderElectionService.getServerId(),
                    "zxid", leaderElectionService.getZxid(),
                    "type", "客户端投票的信息"
            ));
            session.sendMessage(new TextMessage(message));
        } catch (Exception e) {
            System.err.println("⚠️ Error sending vote: " + e.getMessage());
        }
    }

    private void closeSession(String url) {
        WebSocketSession session = activeSessions.get(url);
        if (session != null) {
            try {
                session.close();
            } catch (Exception e) {
                System.err.println("⚠️ Error closing session for " + url);
            }
            activeSessions.remove(url);
        }
    }
}
